package cn.doitedu.demo3;

import cn.doitedu.beans.EventBean;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.List;

public class RuleCalculator_02 implements RuleCalculator {
    ValueState<List<EventBean>> state;
    JSONObject message ;

    String rule_id;

    @Override
    public void init(RuntimeContext runtimeContext  ,String ruleParamJson) {

        rule_id = "rule-02";

        state = runtimeContext.getState(new ValueStateDescriptor<List<EventBean>>(rule_id+"-lst", TypeInformation.of(new TypeHint<List<EventBean>>() {
        })));

        message = new JSONObject();
        message.put("rule_id",rule_id);
    }

    @Override
    public void calc(EventBean eventBean, Collector<String> out) throws IOException {
        String eventId = eventBean.getEvent_id();
        long userId = eventBean.getUser_id();
        long actionTime = eventBean.getAction_time();

        if("QD".equals(eventId)){

            // 判断加购事件列表中的  商品平均价格  和 最小价格
            List<EventBean> list = state.value();
            if(list != null && list.size()>=3 ){
                int cnt = 0;
                double priceSum = 0;
                double minPrice = Double.MAX_VALUE;
                for (EventBean bean : list) {
                    double price = Double.parseDouble(bean.getProperties().get("item_price"));
                    priceSum += price;
                    cnt++;
                    if(price < minPrice){
                        minPrice = price;
                    }
                }

                if(priceSum/cnt > 300 && minPrice > 50){
                    message.put("user_id",userId);
                    message.put("hit_time",actionTime);
                    out.collect(message.toJSONString());
                }


            }



        }


        List<EventBean> list = state.value();
        if("add_cart".equals(eventId)){
            if(list== null){
                state.update(Lists.newArrayList(eventBean));
            }else if(list.size()<3){
                list.add(eventBean);
            }else{
                list.remove(0);
                list.add(eventBean);

            }

        }
    }

    @Override
    public String getRuleId() {
        return this.rule_id;
    }

}
